package kafka

import (
	"encoding/json"
	"fmt"
	"github.com/Shopify/sarama"
	"go.uber.org/zap"
	"logtransfer/es"
)

var (
	consumer sarama.Consumer
)

func Init(addrs []string) (err error) {
	consumer, err = sarama.NewConsumer(addrs, nil)
	if err != nil {
		err = fmt.Errorf("创建消费者失败,err=", err)

	}
	return

}

func WatchData(topic string) (err error) {

	partitionList, err := consumer.Partitions(topic) // 根据topic取到所有的分区

	if err != nil {

		return err

	}
	for partition := range partitionList { // 遍历所有的分区
		// 针对每个分区创建一个对应的分区消费者
		pc, err := consumer.ConsumePartition(topic, int32(partition), sarama.OffsetNewest)
		if err != nil {

			return err

		}
		// 如果放开下面，则随着宿主函数的退出，消费者也会退出。

		//defer pc.AsyncClose()
		// 异步从每个分区消费信息

		go func(sarama.PartitionConsumer) {

			for msg := range pc.Messages() {
				//fmt.Printf("Partition:%d Offset:%d Key:%s Value:%s\n", msg.Partition, msg.Offset, msg.Key, msg.Value)
				//println(string(msg.Value))

				// 从 kafka获取的字符串可能不是标准的 json 格式

				var v interface{}
				json.Unmarshal(msg.Value, &v)
				if err != nil {
					zap.L().Error("从kafka获取数据反序列化失败，err = ", zap.Error(err))
					continue
				}

				data, ok := v.(map[string]interface{})
				if !ok {

					strs := fmt.Sprintf("类型断言失败，不是标准json, msg = %s", msg.Value)
					zap.L().Error(strs)
					continue
				}

				//var m1 map[string]interface{}
				//err = json.Unmarshal(msg.Value, &m1)
				//if err != nil {
				//	zap.L().Error("从kafka获取数据反序列化失败，err = ", zap.Error(err))
				//	return
				//}
				// 日志丢入通道
				es.PutLogData(data)

			}

		}(pc)

	}
	return err
}
